package com.mjf.spark.day10

import kafka.serializer.StringDecoder
import org.apache.kafka.clients.consumer.ConsumerConfig
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka.KafkaUtils
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
 * 通过DirectAPI连接Kafka数据源，获取数据
 *    自动的维护偏移量，偏移量维护在checkpoint中
 *    目前我们这个版本，只是指定了检查点，只会将offset放到检查点中，但是并没有从检查点中取，会存在消息丢失
 */
object SparkStreaming05_DirectAPI_Auto01 {
  def main(args: Array[String]): Unit = {

    val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkStreaming")

    val ssc: StreamingContext = new StreamingContext(conf, Seconds(3))

    // 设置检查点目录
    ssc.checkpoint("D:\\code\\spark\\cp")

    // 准备Kafka参数
    val kafkaParams: Map[String, String] = Map[String, String](
      ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -> "hadoop102:9092,hadoop103:9092,hadoop104:9092",
      ConsumerConfig.GROUP_ID_CONFIG -> "bigdata"
    )

    // 连接Kafka，创建DStream
    val kafkaDStream: InputDStream[(String, String)] = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](
      ssc,
      kafkaParams,
      Set("bigdata-mjf")
    )

    // 获取Kafka中的数据，我们只需要v的部分
    val lineDS: DStream[String] = kafkaDStream.map(_._2)

    // 扁平化
    val flatMapDS: DStream[String] = lineDS.flatMap(_.split(" "))

    // 结构转换 进行计数
    val mapDS: DStream[(String, Int)] = flatMapDS.map((_, 1))

    // 聚合操作
    val reduceDS: DStream[(String, Int)] = mapDS.reduceByKey(_ + _)

    // 打印输出
    reduceDS.print()

    // 启动采集器
    ssc.start()

    // 等待采集结束之后，终止程序
    ssc.awaitTermination()

  }
}
